import sys
sys.path.append('C:\Program Files\Tinysoft\Analyse.NET')
import TSLPy3 as ts
import info
import tsbytestostr as tr
import pandas as pd
import datetime
import os
import numpy as np
# import uqer_enterprise
import re
import math
import paramiko
import random

class GetExternalDataSource:
    """
    get external data source
    """
    def __init__(self):
        """
        载入常用数据
        """
        self.root_path = info.DATAPATH.dataRootPath  # 路径信息
        self.server_root_path = info.DATAPATH.server_root_path  # 服务器路径
        self.t, self.sftp =self.get_server()  # 与服务器建立连接


    def get_server(self, hostname='172.16.0.11', username='root', password='111111', port=22):
        """
        与服务器建立连接

        :param hostname: 服务器地址
        :param username: 用户名
        :param password: 密码
        :param port: 接口
        :return:
        """
        t = paramiko.Transport((hostname, port))
        t.connect(username=username, password=password)
        sftp = paramiko.SFTPClient.from_transport(t)
        return t, sftp

    def mkdir_p(self, sftp, remote, is_dir=False):
        """
        emulates mkdir_p if required.
        sftp - is a valid sftp object
        remote - remote path to create.
        """
        dirs_ = []
        if is_dir:
            dir_ = remote
        else:
            dir_, basename = os.path.split(remote)
        while len(dir_) > 1:
            dirs_.append(dir_)
            dir_, _ = os.path.split(dir_)

        if len(dir_) == 1 and not dir_.startswith("/"):
            dirs_.append(dir_)  # For a remote path like y/x.txt

        while len(dirs_):
            dir_ = dirs_.pop()
            try:
                sftp.stat(dir_)
            except:
                print("making ... dir", dir_)
                sftp.mkdir(dir_)

    def put_file(self, local_path, remote_path):
        """
        向服务器传输文件
        :param local_path: 本机地址
        :param remote_path: 远程地址
        :return:
        """
        self.mkdir_p(self.sftp, remote_path)
        self.sftp.put(local_path, remote_path)

    def getTradingCalendar(self, endt='2022-12-31', begt='2010-01-01'):
        """
        :param endt: trading start date
        :param begt: trading end date
        :return:

        从天软获取交易日历
        """

        tsstr = """
            begt:=strtodate('""" + begt + """'); 
            endt:=strtodate('""" + endt + """'); 
            return DateToStr(spec(specdate(nday3(tradedays(begt,endt),sp_time()),endt),'SH000001')); 
        """
        raw_data = tr.tsbytestostr(ts.RemoteExecute(tsstr, {}))[1]
        df = pd.DataFrame(raw_data)
        df.columns = ['TradingDate']
        assert not df.empty
        if os.path.exists(info.DATAPATH.dataRootPath + '\\calendar'):
            pass
        else:
            os.mkdir(info.DATAPATH.dataRootPath + '\\calendar')
        df.to_pickle(info.DATAPATH.dataRootPath + '\\' + 'calendar' + '\\' + 'tradingCalendar.pkl')
        local_path = info.DATAPATH.dataRootPath + '\\' + 'calendar' + '\\' + 'tradingCalendar.pkl'
        remote_path = info.DATAPATH.server_root_path + '/' + 'calendar' + '/' + 'tradingCalendar.pkl'
        self.put_file(local_path, remote_path)
        print('Downloaded calendar to {}'.format(df['TradingDate'].iloc[-1]))

    def getProductNames(self, endt: str):
        """
        :param endt: lastest date
        :return:

        从天软获取截止到 endt 的全部商品名称
        """
        print('product names start to download.')
        tsstr = """
        SetSysParam("CurrentDate", strtodate('""" + endt + """'));
        SetSysParam("Cycle","日线");
        SetSysParam("bRate",1);
        SetSysParam("RateDay",0);
        SetSysParam("Precision",2);
        SetSysParam("profiler",0);
        SetSysParam("ReportMode",0);
        SetSysParam("EmptyMode",0);
        SetSysParam("CalcCTRLWord",0);
        SetSysParam("LanguageID",0);
        return 
        Query("期货品种代码","",True,"","代码",DefaultStockID(),
        "名称",CurrentStockName());
        """
        raw_data = tr.tsbytestostr(ts.RemoteExecute(tsstr, {}))[1]
        df = pd.DataFrame(raw_data)
        df.to_pickle(info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ProductNames.pkl')
        df.to_csv(info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ProductNames.csv')

        local_path = info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ProductNames.pkl'
        remote_path = info.DATAPATH.server_root_path + '/' + 'future_basic_data' + '/' + 'ProductNames.pkl'
        self.put_file(local_path, remote_path)

        print('Downloaded product names')

    def getContractNames(self):
        """
        :return:
        根据商品名称获取历史上全部合约
        """
        ProductNames = pd.read_pickle(info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ProductNames.pkl')
        names = ProductNames['名称'].tolist()
        FutureContractNames_list = []
        for name_for_query in names:
            tsstr = """
                    return getbkall('""" + name_for_query + """');
            """
            raw_data = tr.tsbytestostr(ts.RemoteExecute(tsstr, {}))[1]
            df = pd.DataFrame(raw_data)
            if df.empty:
                print('ERROR: {} has not data.'.format(name_for_query))
            else:
                df.columns = ['ContractName']
                df = self.getContractBasicInfo(name_for_query, df.ContractName.tolist())
                df['ProductName'] = name_for_query
                df['ProductCode'] = ProductNames.loc[ProductNames['名称']==name_for_query, '代码'].values[0]

                FutureContractNames_list.append(df)
        FutureContractNames_df = pd.concat(FutureContractNames_list)
        FutureContractNames_df.to_pickle(info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ContractInfo.pkl')
        FutureContractNames_df.to_csv(
            info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ContractInfo.csv')
        local_path = info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ContractInfo.pkl'
        remote_path = info.DATAPATH.server_root_path + '/' + 'future_basic_data' + '/' + 'ContractInfo.pkl'
        self.put_file(local_path, remote_path)


    def getContractBasicInfo(self, productName: str, contractNames: str):
        """
        :param productName: 商品名称
        :param contractNames: 合约名称
        :return:

        从天软上下载合约基本信息:

        "变动日"
        "合约乘数单位"
        "合约乘数"
        "最后交易日"
        "最小变动价位"
        "报价单位"
        "每日价格最大波动下限(%)"
        "每日价格最大波动上限(%)"
        "开仓手续费(交易额万分之)"
        "开仓手续费(元/手)"
        "平仓手续费(交易额万分之)"
        "平仓手续费(元/手)"
        "短线开仓手续费(交易额万分之)"
        "短线开仓手续费(元/手)"
        "短线平仓手续费(交易额万分之)"
        "短线平仓手续费(元/手)"
        "上市地"

        """
        contractNames_for_query = ';'.join(contractNames)
        tsstr = """
        return
        Query('""" + productName + """', '""" + contractNames_for_query + """',True,"","代码",DefaultStockID(),
        "变动日",base(703002, 0),
        "合约乘数单位",base(703008),
        "合约乘数",base(703007),
        "最后交易日",base(703018),
        "最小变动价位",base(703010),
        "报价单位",base(703009),
        "每日价格最大波动下限(%)", base(703011),
        "每日价格最大波动上限(%)", base(703012),
        "开仓手续费(交易额万分之)",base(704006),
        "开仓手续费(元/手)",base(704007),
        "平仓手续费(交易额万分之)",base(704008),
        "平仓手续费(元/手)",base(704009),
        "短线开仓手续费(交易额万分之)",base(704010),
        "短线开仓手续费(元/手)",base(704011),
        "短线平仓手续费(交易额万分之)",base(704012),
        "短线平仓手续费(元/手)",base(704013),
        "上市地",base(703027)
        );
        """
        raw_data = tr.tsbytestostr(ts.RemoteExecute(tsstr, {}))[1]
        df = pd.DataFrame(raw_data)
        return df


    def getBarData(self, product_names=['苹果']):
        """
        :param contract_name: 合约名称
        :param begt: 开始日
        :param endt: 结束日
        :return:

        从天软上获取数据:
        high: 最高价
        open: 开盘价
        close: 收盘价
        low: 最低价
        yclose: 昨收
        vol: 成交量, 合约数量
        amount: 成交金额
        cjbs: 周期内的持仓的变动量，日线的话表示当日持仓量
        syl1: 如果是日线，表示当日结算价
        syl2: 昨日结算价
        """
        FutureContractInfo = pd.read_pickle(
            info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ContractInfo.pkl')
        # product_names = FutureContractInfo['ProductName'].unique().tolist()
        calendar = pd.read_pickle(info.DATAPATH.dataRootPath + '\\' + 'calendar' + '\\' + 'tradingCalendar.pkl')
        calendar.index = calendar['TradingDate'].tolist()
        first_trade_date = calendar.index[0]
        for product_name in product_names:
            FutureContractInfo_tmp = FutureContractInfo[FutureContractInfo.ProductName==product_name]
            for index, row in FutureContractInfo_tmp.iterrows():
                contract_name = row['代码']
                begt = str(row['变动日'])
                begt = begt[:4] + '-' + begt[4:6] + '-' + begt[6:]
                endt = str(row['最后交易日'])
                endt = endt[:4] + '-' + endt[4:6] + '-' + endt[6:]
                if begt>first_trade_date:
                    # self.getDailyBarData(contract_name, product_name, begt, endt) # 日线
                    self.getHFBarData(contract_name, product_name, begt, endt)  # 高频

    def getDailyBarData(self, contract_name: str, product_name: str, begt: str, endt: str):
        """
        :param contract_name: 合约名称
        :param begt: 开始日
        :param endt: 结束日
        :return:

        日线参数:
        high: 最高价
        open: 开盘价
        close: 收盘价
        low: 最低价
        yclose: 昨收
        vol: 成交量, 合约数量
        amount: 成交金额
        cjbs: 周期内的持仓的变动量，日线的话表示当日持仓量
        syl1: 如果是日线，表示当日结算价
        syl2: 昨日结算价
        """
        print('{} {} {}'.format(contract_name, begt, endt))
        tsstr = """
        setsysparam(pn_cycle(),cy_day());
        SetSysParam(Pn_Rate(),0);
        t:= select ['StockID'],datetostr(['date']) as 'date',['high'],
             ['open'],['close'],['low'],['yclose'],['vol'],['amount'],['cjbs'],['syl1'],['syl2']
           from Markettable datekey strtodate('""" + begt + """') to strtodate('""" + endt + """') of array('""" + contract_name + """') end;
        return t;
        """
        raw_data = tr.tsbytestostr(ts.RemoteExecute(tsstr, {}))[1]
        df = pd.DataFrame(raw_data)
        if not os.path.exists(self.root_path + '\\' + 'future_trade_data' + '\\' + 'dailyBar' + '\\' + product_name):
            os.mkdir(self.root_path + '\\' + 'future_trade_data' + '\\' + 'dailyBar' + '\\' + product_name)
        if not df.empty:
            df.to_pickle(self.root_path + '\\' + 'future_trade_data' + '\\' + 'dailyBar' + '\\' + product_name + '\\' + contract_name + '.pkl')
            local_path = self.root_path + '\\' + 'future_trade_data' + '\\' + 'dailyBar' + '\\' + product_name + '\\' + contract_name + '.pkl'
            remote_path = self.server_root_path + '/' + 'future_trade_data' + '/' + 'dailyBar' + '/' + product_name + '/' + contract_name + '.pkl'
            self.put_file(local_path, remote_path)
        else:
            print('{} has no data'.format(contract_name))

    def sftp_exists(self, sftp, path):
        """
        :param sftp:
        :param path:
        :return:

        检查远程地址是否存在
        """
        try:
            sftp.stat(path)
            return True
        except FileNotFoundError:
            return False

    def getHFBarData(self, contract_name: str, product_name: str, begt: str, endt: str):
        """
        :param contract_name: 合约名称
        :param begt: 开始日
        :param endt: 结束日
        :return:

        高频参数:
        high: 最高价
        open: 开盘价
        close: 收盘价
        low: 最低价
        yclose: 昨收
        vol: 成交量, 合约数量
        amount: 成交金额
        cjbs: 周期内的持仓的变动量，日线的话表示当日持仓量
        syl1: 如果是日线，表示当日结算价
        syl2: 昨日结算价
        """
        calendar = pd.read_pickle(info.DATAPATH.dataRootPath + '\\' + 'calendar' + '\\' + 'tradingCalendar.pkl')
        calendar.index = calendar['TradingDate'].tolist()
        tradingPeriod = calendar[begt:endt].index.tolist()
        print(contract_name, product_name, begt, endt)
        if not os.path.exists(self.root_path + '\\' + 'future_trade_data' + '\\' + 'hfBar' + '\\' + product_name):
            os.mkdir(self.root_path + '\\' + 'future_trade_data' + '\\' + 'hfBar' + '\\' + product_name)


        for tradingDate in tradingPeriod:
            remote_path = self.server_root_path + '/' + 'future_trade_data' + '/' + 'hfBar' + '/' + product_name + '/' + contract_name + '/' + tradingDate + '.pkl'
            remote_exist = self.sftp_exists(self.sftp, remote_path)
            if not remote_exist:
                tsstr = """
                setsysparam(pn_cycle(),cy_halfs());
                SetSysParam(Pn_Rate(),0);
                t:= select ['StockID'], FormatDateTime('yyyy-mm-dd hh:nn:ss:zzz', ['date']) as 'date',['high'],
                ['open'],['close'],['low'],['yclose'],['vol'],['amount'],['cjbs'],['syl1'],['syl2']
                from Markettable 
                datekey strtodate('""" + tradingDate + """') to strtodate('""" + tradingDate + """')+0.99999 of array('""" + contract_name + """') end;
                return t;
                """
                daily_bar_data = pd.read_pickle(self.root_path + '\\' + 'future_trade_data' + '\\' + 'dailyBar' + '\\' + product_name + '\\' + contract_name + '.pkl')
                raw_data = tr.tsbytestostr(ts.RemoteExecute(tsstr, {}))[1]
                df = pd.DataFrame(raw_data)
                if tradingDate in daily_bar_data.date.tolist():
                    df['syl1'] = daily_bar_data.loc[daily_bar_data.date==tradingDate, 'syl1'].values[0]

                # if not os.path.exists(self.root_path + '\\' + 'future_trade_data' + '\\' + 'hfBar' + '\\' + product_name + '\\' + contract_name):
                #     os.mkdir(self.root_path + '\\' + 'future_trade_data' + '\\' + 'hfBar' + '\\' + product_name + '\\' + contract_name)
                    if not df.empty:
                        df.to_pickle('tmp\\' + tradingDate + '.pkl')
                        local_path = 'tmp\\' + tradingDate + '.pkl'
                        self.put_file(local_path, remote_path)
                        os.remove(local_path)
                    else:
                        print('{} {} empty.'.format(tradingDate, contract_name))
                else:
                    print('{}: {} no daily data.'.format(tradingDate, contract_name))
            else:
                print('{} {} exists on remote.'.format(contract_name, tradingDate))


    def getZL(self, product_names=None):
        """
        :param product_names:
        :return:

        计算主力合约，计算逻辑:
        1. T-1日收盘计算持仓和成交量最大的合约， 最大者为主力合约
        2. 若主力合约发生变化，则在T日换合约
        3. 合约只向后切换
        4. 合约最后一个交易日必换新合约，若新合约不满足持仓和成交量同时最大, 则选最近的合约
        """
        if product_names == None:
            product_names = os.listdir(self.root_path + '\\' + 'future_trade_data' + '\\' + 'dailyBar')
        contract_info = pd.read_pickle(self.root_path + '\\' + 'future_basic_data' + '\\' + 'ContractInfo.pkl')
        contract_info = contract_info.drop_duplicates()
        contract_info['last_trade_date'] = [str(x)[:4] + '-' + str(x)[4:6] + '-' + str(x)[6:] for x in contract_info['最后交易日'].tolist()]

        ZL_dict = {}
        for product_name in product_names:
            print('{} start get ZL data.'.format(product_name))
            history_contract_filenames = os.listdir(self.root_path + '\\' + 'future_trade_data' + '\\' + 'dailyBar' + '\\' + product_name)
            contract_vol_dict = {}
            contract_cjbs_dict = {}
            for contract_filename in history_contract_filenames:
                contract_trade_info = pd.read_pickle(self.root_path + '\\' + 'future_trade_data' + '\\' + 'dailyBar' + '\\' + product_name + '\\' + contract_filename)
                contract_vol_dict[contract_filename[:-4]] = pd.Series(contract_trade_info['vol'].tolist(), index=contract_trade_info['date'].tolist())
                contract_cjbs_dict[contract_filename[:-4]] = pd.Series(contract_trade_info['cjbs'].tolist(), index=contract_trade_info['date'].tolist())

            contract_vol_raw = pd.DataFrame(contract_vol_dict)
            contract_cjbs_raw = pd.DataFrame(contract_cjbs_dict)

            contract_vol_info_his = contract_vol_raw.shift(1)
            contract_cjbs_info_his = contract_cjbs_raw.shift(1)
            contract_vol_info_his[contract_vol_raw.isnull()] = np.NaN
            contract_cjbs_info_his[contract_cjbs_raw.isnull()] = np.NaN
            contract_vol_info_his_max_mask = contract_vol_info_his.eq(contract_vol_info_his.max(axis=1), axis=0)
            contract_cjbs_info_his_max_mask = contract_cjbs_info_his.eq(contract_cjbs_info_his.max(axis=1), axis=0)
            vol_cjbs_info = contract_vol_info_his_max_mask & contract_cjbs_info_his_max_mask
            vol_cjbs_info[contract_vol_info_his.isnull() | contract_vol_info_his.isnull()] = False
            vol_cjbs_info[contract_vol_raw.isnull() | contract_vol_raw.isnull()] = np.NaN
            vol_cjbs_info.dropna(how='all', inplace=True)
            vol_cjbs_contract_name_dict = {}
            last_int_date = 0
            for index, row in vol_cjbs_info.iterrows():
                contract_waitinglist_info = row.to_frame(name='waitinglist')

                contract_waitinglist_info['last_trade_date'] = contract_info.loc[np.in1d(contract_info['代码'].tolist(), row.index.tolist()), 'last_trade_date'].tolist()
                contract_waitinglist_info.loc[contract_waitinglist_info.last_trade_date==index, 'waitinglist'] = np.NaN


                contract_waitinglist = contract_waitinglist_info.waitinglist.dropna()
                if contract_waitinglist.sum() > 0:
                    contract_name = contract_waitinglist.index[contract_waitinglist.tolist()].values[0]
                else:
                    if len(contract_waitinglist) == 0:
                        print(row)
                        contract_name = row.dropna().index[0]
                    else:
                        contract_name = contract_waitinglist.index[0]
                if str(contract_name)!='nan':
                    int_date = int(re.findall(r'\d+', contract_name)[0])
                    if int_date>=last_int_date:
                        vol_cjbs_contract_name_dict[index] = contract_name
                        last_int_date = int_date
                    else:
                        vol_cjbs_contract_name_dict[index] = np.NaN
                else:
                    vol_cjbs_contract_name_dict[index] = np.NaN

            vol_cjbs_contract_name = pd.Series(vol_cjbs_contract_name_dict)

            vol_cjbs_contract_name.fillna(method='ffill', inplace=True)
            ZL_dict[product_name] = vol_cjbs_contract_name.dropna(how='all')
        ZL_df = pd.DataFrame(ZL_dict)
        ZL_df.to_pickle(info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ZL.pkl')

        local_path = 'tmp\\' + 'ZL' + '.pkl'
        ZL_df.to_pickle(local_path)
        remote_path = self.server_root_path + '/' + 'future_basic_data' + '/' + 'ZL.pkl'
        self.put_file(local_path, remote_path)
        os.remove(local_path)

    def getPriceRate(self):
        """
        :return:
        获得后复权因子

        后复权因子(T) = T-1日旧合约收盘价 - T-1日新合约收盘价
        """
        # 载入交易日历
        calendar = pd.read_pickle(info.DATAPATH.dataRootPath + '\\' + 'calendar' + '\\' + 'tradingCalendar.pkl')
        calendar.index = calendar['TradingDate'].tolist()
        calendar['yesterday'] = calendar['TradingDate'].shift(1).tolist()
        # 读取主力合约列表
        ZL_df = pd.read_pickle(info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ZL.pkl')
        ZL_df = ZL_df.loc[:calendar.index[-1], :]
        if '晚籼稻' in ZL_df.columns.tolist():
            ZL_df.drop(columns = ['晚籼稻'], inplace=True)
        priceRate_dict = {}
        for col in ZL_df.columns.tolist():
            print(col)
            priceRate_tmp_dict = {}
            ZL_tmp_tradingDates = ZL_df[col].dropna().index.tolist()
            for ZL_tmp_tradingDate in ZL_tmp_tradingDates:
                yesterday = calendar.loc[ZL_tmp_tradingDate, 'yesterday']
                if yesterday in ZL_tmp_tradingDates:
                    yesterday_contract_name = ZL_df.loc[yesterday, col]
                    today_contract_name = ZL_df.loc[ZL_tmp_tradingDate, col]
                    if yesterday_contract_name==today_contract_name:
                        priceRate_tmp_dict[ZL_tmp_tradingDate] = np.NaN
                    else:

                        yesterday_contract_trade_info = pd.read_pickle(info.DATAPATH.dataRootPath + '\\' +
                                                                       'future_trade_data' + '\\' +
                                                                       'dailyBar' + '\\' +
                                                                       col + '\\' +
                                                                       yesterday_contract_name + '.pkl')
                        today_contract_trade_info = pd.read_pickle(info.DATAPATH.dataRootPath + '\\' +
                                                                       'future_trade_data' + '\\' +
                                                                       'dailyBar' + '\\' +
                                                                   col + '\\' +
                                                                   today_contract_name + '.pkl')
                        yesterday_contract_trade_info.index = yesterday_contract_trade_info.date.tolist()
                        today_contract_trade_info.index = today_contract_trade_info.date.tolist()

                        priceRate_tmp_dict[ZL_tmp_tradingDate] = yesterday_contract_trade_info.loc[yesterday, 'close'] \
                                                                 - today_contract_trade_info.loc[yesterday, 'close']
                else:
                    priceRate_tmp_dict[ZL_tmp_tradingDate] = np.NaN
            priceRate_tmp_series = pd.Series(priceRate_tmp_dict)
            priceRate_tmp_series.fillna(0, inplace=True)
            priceRate_dict[col] = priceRate_tmp_series.cumsum()
        priceRate_df = pd.DataFrame(priceRate_dict)
        priceRate_df.index = pd.to_datetime(priceRate_df.index.tolist(), format='%Y-%m-%d').tolist()
        priceRate_df.to_pickle(info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'PriceRate.pkl')
        local_path = 'tmp\\' + 'PriceRate.pkl'
        priceRate_df.to_pickle('tmp\\' + 'PriceRate.pkl')
        remote_path = self.server_root_path + '/' + 'future_basic_data' + '/' + 'PriceRate.pkl'
        self.put_file(local_path, remote_path)
        os.remove(local_path)



    def update_data(self, endt):
        # calendar
        self.getTradingCalendar(endt)
        # product_info
        self.getProductNames(endt)
        # contract_info
        self.getContractNames()



if __name__ == '__main__':
    endt = datetime.date.today().strftime("%Y-%m-%d")
    endt = '2022-06-24'
    a = GetExternalDataSource()
    # a.getTradingCalendar(endt)
    # a.getProductNames(endt)
    # a.getContractNames()
    # total_product_names = ['中证500指数', '沪深300指数']
    df = pd.read_pickle(info.DATAPATH.dataRootPath + '\\' + 'future_basic_data' + '\\' + 'ProductNames.pkl')
    product_names = df['名称'].tolist()
    # a.getBarData(product_names)
    # a.getZL()
    # a.getPriceRate()
    product_names = ['沥青']
    a.getBarData(product_names)
    print()